#include<stdio.h>
#include<stdlib.h>
#include<sys/socket.h>
#include<sys/types.h>
#include<arpa/inet.h>
#include<unistd.h>
#include<memory.h>
#include<time.h>
#include<string.h>
#include<sys/wait.h>
#include<signal.h>
#include<time.h>
#include<sys/time.h>
#include<pthread.h>
#include<semaphore.h>
#define SERV_PORT 8888
#define SOCK_COUNT 30
#define MAXSIZE 1000
#define BUF_COUNT 10
#define MSG_FILENAME 1
#define MSG_CONTINUE 2
#define MSG_ACK 3
#define MSG_DONE 4
#define MSG_EXCEPTION 5
//消息结构体
struct msg
{
    int  type;
    int len;
    char data[];
};


int main(int argc,char *argv[])
{
    int sockfd,new_fd;
    struct sockaddr_in servaddr,cliaddr;
    int sin_size,numbytes;
    pid_t pid;
    struct timeval start;
    struct timeval end;
    char strptr[16];
    //创建socket
    if((sockfd = socket(AF_INET,SOCK_STREAM,0))<0)
    {
        perror("socket error");
        exit(-1);
    }
    //初始化socket结构体
    memset(&servaddr,0x00,sizeof(servaddr));
    memset(&cliaddr,0x00,sizeof(cliaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(SERV_PORT);
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    //避免僵死进程
    void sig_child(int signo)
    {
        pid_t pid;
        int stat;
        while((pid = waitpid(-1,&stat,WNOHANG))>0)
        {
            printf("chlid %d terminated\n",pid);
        }
        return;
    }
    //绑定套接口
    if(bind(sockfd,(struct sockaddr *)&servaddr,sizeof(struct sockaddr)) == -1)
    {
        perror("bind error");
        exit(-1);
    }
    //创建监听套接口
    if(listen(sockfd,SOCK_COUNT) == -1)
    {
        perror("listen error");
        exit(-1);
    }
    //等待连接
    //printf("0-1\n");
    while(1)
    {
        sin_size = sizeof(struct sockaddr_in) ;
        if((new_fd = accept(sockfd,(struct sockaddr *)&cliaddr,(socklen_t *)&sin_size)) == -1)
        {
            perror("accept error");
            exit(-1);
        }
        //创建子进程
        if((pid=fork()) == 0)
        {
            int head=0;//队头指针
            int rear=0;//队尾指针
            struct msg *buf_ptr[BUF_COUNT];//缓冲区队列
            sem_t count;//已经使用的缓冲区
            sem_t empty_count;//未使用的缓冲区
            int i;
            close(sockfd);
            struct msg *sm,*rm;
            pthread_t ptid1,ptid2;
            rm=(struct msg *)malloc(MAXSIZE+sizeof(struct msg));
            sm=(struct msg *)malloc(MAXSIZE+sizeof(struct msg));
            //初始化确认信息
            sm->type = MSG_ACK;
            sm->len = 0;
            gettimeofday(&start,NULL);
            //将消息结构体，拷贝进缓冲区
            void cpy_sm_to_buf(struct msg *sm)
            {

                sem_wait(&empty_count);
                memset(buf_ptr[rear],0x00,MAXSIZE);
                memcpy(buf_ptr[rear],sm,sizeof(struct msg)+sm->len);
                //更新队列状态
                rear = (rear+1)%BUF_COUNT;
                //控制线程2退出
                sem_post(&count);
            }
            //生产者，接收数据并放入缓存区
            void* producer(void *arg)
            {
                int th_new_fd= *(int *)arg;
                int  producer_flag = MAXSIZE;
                while( producer_flag != 0)
                {
                    if((numbytes = recv(th_new_fd,(void *)rm,sizeof(struct msg),0)) == -1)
                    {
                        perror("recv error");
                        exit(-1);
                    }
                    if(numbytes == 0)//判断客户端是否断开
                    {
                        printf("客户端已断开！\n");
                        exit(-1);
                    }
                    producer_flag = rm->len;

                    if((rm->type == MSG_DONE)||(rm->type == MSG_EXCEPTION))
                    {
                        cpy_sm_to_buf(rm);
                        break;
                    }
                    memset(rm->data,0x00,MAXSIZE);
                    if((numbytes = recv(th_new_fd,rm->data,rm->len,0)) == -1)
                    {
                        perror("recv error");
                        exit(-1);
                    }
                    cpy_sm_to_buf(rm);
                }
                free(rm);
                pthread_exit(NULL);
            }
            //消费者，将缓存区的内容写入文件
            void* consumer()
            {
                FILE *fp;
                int consume_flag = MAXSIZE;
                while(consume_flag != 0)
                {
                    sem_wait(&count);
                    consume_flag = buf_ptr[head]->len;
                    if(buf_ptr[head]->type == MSG_FILENAME)
                    {
                        if(buf_ptr[head]->len>256)
                        {
                            printf("文件名过长！\n");
                            exit(-1);
                        }
                        char file_name[256];
                        memset(file_name,0x00,256);
                        memcpy(file_name,buf_ptr[head]->data,buf_ptr[head]->len);
                        if((fp = fopen(file_name,"w+")) == NULL)
                        {
                            perror("fopen error");
                            exit(-1);
                        }
                    }
                    else if(buf_ptr[head]->type == MSG_CONTINUE)
                    {
                        if(fwrite(buf_ptr[head]->data,1,buf_ptr[head]->len,fp)<0)
                        {
                            printf("fwrite error\n");
                            exit(-1);
                        }
                    }
                    else if(buf_ptr[head]->type == MSG_DONE)
                    {
                        printf("send MSG_DONE!\n");
                        break;
                    }
                    else if(buf_ptr[head]->type == MSG_EXCEPTION)
                    {
                        printf("read the file failed！%d\n",buf_ptr[head]->type);
                        exit(-1);
                    }
                    else
                    {
                        printf("文件类型错误!rm->type=%d\n",buf_ptr[head]->type);
                        exit(-1);
                    }
                    //更新队列状态
                    head = (head+1)%BUF_COUNT;
                    sem_post(&empty_count);
                }
                fclose(fp);
                pthread_exit(NULL);
            }
            //为存放数据的缓冲去分配内存
            for(i=0; i<BUF_COUNT; i++)
            {
                buf_ptr[i]=(struct msg *)malloc(MAXSIZE+sizeof(struct msg));
            }
            if(sem_init(&count,0,0) == -1)
            {
                perror("sem_init error");
                exit(-1);
            }
            if(sem_init(&empty_count,0,BUF_COUNT) == -1)
            {
                perror("sem_init error");
                exit(-1);
            }
            gettimeofday(&start,NULL);
            pthread_create(&ptid1,NULL,producer,&new_fd);
            pthread_create(&ptid2,NULL,consumer,NULL);
            pthread_join(ptid1,NULL);
            pthread_join(ptid2,NULL);
            gettimeofday(&end,NULL);
            sem_destroy(&count);
            sem_destroy(&empty_count);

            printf("the time of receiving the file is %ld\n",(end.tv_sec - start.tv_sec)*1000000+(end.tv_usec - start.tv_usec));
            close(new_fd);
            free(sm);
            //释放缓冲队列
            for(i=0; i<BUF_COUNT; i++)
            {
                free(buf_ptr[i]);
            }
            exit(0);
        }
        //通过信号，防止僵尸进程
        signal(SIGCHLD,sig_child);
        //服务器端打印客户端的网址、端口号
        if(inet_ntop(AF_INET,&cliaddr.sin_addr,strptr,(socklen_t)sizeof(strptr)) == NULL)
        {
            perror("convert error");
            exit(-1);
        }
        printf("connect from %s,port is %d,pid is %d\n",strptr,ntohs(cliaddr.sin_port),pid);
        close(new_fd);
    }
    close(sockfd);
    return 0;
}
